package com.tianmu.connector.client.eventhandler.rocketmq;

import com.tianmu.connector.server.BinlogConnectorException;
import com.tianmu.connector.server.event.EventEntity;
import com.tianmu.connector.server.event.handler.IEventHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * rocketMQ 处理监听后的变更消息
 * @author tianmu
 */
@Slf4j
@Component
public class RocketMqEventHandler implements IEventHandler {
    @Resource
    private RocketMQTemplate rocketMqTemplate;

    private static final String TOPIC = "data_connector_topic";
    @Override
    public void process(EventEntity eventEntity) throws BinlogConnectorException {
        if (eventEntity == null) {
            return;
        }
        String jsonFormatData = eventEntity.getJsonFormatData();
        if (StringUtils.isBlank(jsonFormatData)) {
            return;
        }
        SendResult sendResult = rocketMqTemplate.syncSend(TOPIC, jsonFormatData);
        log.info("sync send message success.sendResult:{},params:{}", sendResult, jsonFormatData);
    }
}
